package org.example.rabbitmq.rpc;

import com.rabbitmq.client.*;

public class RPCServer {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    private static int fib(int n) {
        if (n == 0) return 0;
        if (n == 1) return 1;
        return fib(n - 1) + fib(n - 2);
    }

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.0.228");
        factory.setUsername("mq");
        factory.setPassword("123456");

        // 创建一个新的连接到RabbitMQ服务器
        Connection connection = factory.newConnection();
        // 通过连接创建一个新的通道
        Channel channel = connection.createChannel();
        // 声明一个队列，供RPC请求使用。设置队列的名称，不持久化，不排他，不自动删除，没有额外的参数
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
        // 清空队列中的所有消息，以便处理当前的RPC请求
        channel.queuePurge(RPC_QUEUE_NAME);


        // 设置通道的预取质量，限制未确认消息的数量，以保证处理消息的性能和可靠性
        channel.basicQos(1);


        System.out.println(" [x] Awaiting RPC requests");

        // 定义一个投递回调接口的实现，用于处理接收到的消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            // 构建回复属性，用于设置回复消息的关联ID
            AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                    .Builder()
                    .correlationId(delivery.getProperties().getCorrelationId())
                    .build();

            // 初始化回复消息为空字符串
            String response = "";
            try {
                // 解析接收到的消息体为字符串，并转换为整数
                String message = new String(delivery.getBody(), "UTF-8");
                int n = Integer.parseInt(message);

                // 打印接收到的fibonacci函数参数
                System.out.println(" [.] fib(" + message + ")");
                // 计算fibonacci数列的值并构建回复消息
                response += fib(n);
            } catch (RuntimeException e) {
                // 打印异常信息
                System.out.println(" [.] " + e);
            } finally {
                // 发布回复消息到指定的回复队列
                channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
                // 确认处理完成接收到的消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

        /** 开始消费RPC队列中的消息
         * RPC_QUEUE_NAME：指定要消费的队列名称。
         * false：关闭自动确认，需手动调用 basicAck 确认消息。
         * deliverCallback：定义接收到消息后的处理逻辑。
         * (consumerTag -> {})：消费者被取消时的回调，此处为空实现。
         */
        channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> {}));
    }
}
